//! # 健康检查模块
//!
//! 重新导出 rustcloud-observability 模块的健康检查功能，
//! 为 rustcloud-stream 提供统一的健康检查能力。
//! 
//! 通过这种方式保持向后兼容性，同时使用公共功能模块。

// 重新导出 rustcloud-observability 的健康检查功能
pub use rustcloud_observability::health::*;

// 为流处理定制的健康检查实现
use async_trait::async_trait;
use crate::core::StreamResult;

/// 流处理管道健康检查器
pub struct StreamPipelineHealthChecker {
    /// 管道标识符
    pipeline_id: String,
    /// 管道状态检查函数
    status_checker: Box<dyn Fn() -> bool + Send + Sync>,
}

impl StreamPipelineHealthChecker {
    /// 创建新的流管道健康检查器
    pub fn new(
        pipeline_id: String,
        status_checker: Box<dyn Fn() -> bool + Send + Sync>,
    ) -> Self {
        Self {
            pipeline_id,
            status_checker,
        }
    }

    /// 检查管道是否在运行
    pub fn is_pipeline_running(&self) -> bool {
        (self.status_checker)()
    }
}

#[async_trait]
impl HealthChecker for StreamPipelineHealthChecker {
    async fn check_health(&self) -> ComponentHealth {
        let is_running = self.is_pipeline_running();
        
        ComponentHealth {
            status: if is_running {
                HealthStatus::Healthy
            } else {
                HealthStatus::Critical
            },
            message: format!(
                "流管道 {} {}",
                self.pipeline_id,
                if is_running { "正在运行" } else { "已停止" }
            ),
            details: {
                let mut details = std::collections::HashMap::new();
                details.insert("pipeline_id".to_string(), self.pipeline_id.clone());
                details.insert("running".to_string(), is_running.to_string());
                details
            },
            last_check: chrono::Utc::now(),
        }
    }
}

/// 流处理器健康检查器
pub struct StreamProcessorHealthChecker {
    /// 处理器标识符
    processor_id: String,
    /// 处理器统计信息
    stats: std::sync::Arc<tokio::sync::RwLock<ProcessorStats>>,
}

impl StreamProcessorHealthChecker {
    /// 创建新的流处理器健康检查器
    pub fn new(
        processor_id: String,
        stats: std::sync::Arc<tokio::sync::RwLock<ProcessorStats>>,
    ) -> Self {
        Self {
            processor_id,
            stats,
        }
    }
}

#[async_trait]
impl HealthChecker for StreamProcessorHealthChecker {
    async fn check_health(&self) -> ComponentHealth {
        let stats = self.stats.read().await;
        
        // 计算错误率
        let error_rate = if stats.total_messages > 0 {
            (stats.failed_messages as f64 / stats.total_messages as f64) * 100.0
        } else {
            0.0
        };

        let status = if error_rate > 50.0 {
            HealthStatus::Critical
        } else if error_rate > 10.0 {
            HealthStatus::Warning
        } else {
            HealthStatus::Healthy
        };

        ComponentHealth {
            status,
            message: format!(
                "流处理器 {} 错误率: {:.2}%",
                self.processor_id, error_rate
            ),
            details: {
                let mut details = std::collections::HashMap::new();
                details.insert("processor_id".to_string(), self.processor_id.clone());
                details.insert("total_messages".to_string(), stats.total_messages.to_string());
                details.insert("failed_messages".to_string(), stats.failed_messages.to_string());
                details.insert("error_rate_percent".to_string(), format!("{:.2}", error_rate));
                details
            },
            last_check: chrono::Utc::now(),
        }
    }
}

/// 处理器统计信息
#[derive(Debug, Clone, Default)]
pub struct ProcessorStats {
    /// 总消息数
    pub total_messages: u64,
    /// 成功处理消息数
    pub successful_messages: u64,
    /// 失败处理消息数
    pub failed_messages: u64,
    /// 平均处理时间（毫秒）
    pub avg_processing_time_ms: f64,
}

impl ProcessorStats {
    /// 记录成功处理
    pub fn record_success(&mut self, processing_time_ms: f64) {
        self.total_messages += 1;
        self.successful_messages += 1;
        
        // 更新平均处理时间
        self.avg_processing_time_ms = 
            (self.avg_processing_time_ms * (self.total_messages - 1) as f64 + processing_time_ms) 
            / self.total_messages as f64;
    }

    /// 记录失败处理
    pub fn record_failure(&mut self, processing_time_ms: f64) {
        self.total_messages += 1;
        self.failed_messages += 1;
        
        // 更新平均处理时间
        self.avg_processing_time_ms = 
            (self.avg_processing_time_ms * (self.total_messages - 1) as f64 + processing_time_ms) 
            / self.total_messages as f64;
    }

    /// 获取成功率
    pub fn success_rate(&self) -> f64 {
        if self.total_messages > 0 {
            (self.successful_messages as f64 / self.total_messages as f64) * 100.0
        } else {
            0.0
        }
    }

    /// 获取错误率
    pub fn error_rate(&self) -> f64 {
        if self.total_messages > 0 {
            (self.failed_messages as f64 / self.total_messages as f64) * 100.0
        } else {
            0.0
        }
    }
}

/// 流健康管理器
pub struct StreamHealthManager {
    /// 底层健康检查管理器
    inner: HealthCheckManager,
    /// 流标识符
    stream_id: String,
}

impl StreamHealthManager {
    /// 创建新的流健康管理器
    pub fn new(stream_id: String, config: HealthCheckConfig) -> Self {
        Self {
            inner: HealthCheckManager::new(config),
            stream_id,
        }
    }

    /// 注册流管道健康检查器
    pub fn register_pipeline_checker(
        &mut self,
        pipeline_id: String,
        status_checker: Box<dyn Fn() -> bool + Send + Sync>,
    ) {
        let checker = Box::new(StreamPipelineHealthChecker::new(pipeline_id.clone(), status_checker));
        self.inner.register_checker(format!("pipeline.{}", pipeline_id), checker);
    }

    /// 注册流处理器健康检查器
    pub fn register_processor_checker(
        &mut self,
        processor_id: String,
        stats: std::sync::Arc<tokio::sync::RwLock<ProcessorStats>>,
    ) {
        let checker = Box::new(StreamProcessorHealthChecker::new(processor_id.clone(), stats));
        self.inner.register_checker(format!("processor.{}", processor_id), checker);
    }

    /// 检查所有流组件健康状态
    pub async fn check_stream_health(&self) -> StreamResult<StreamHealthResult> {
        let result = self.inner.check_all().await
            .map_err(|e| crate::error::StreamError::HealthCheck(e.to_string()))?;

        Ok(StreamHealthResult {
            stream_id: self.stream_id.clone(),
            overall_status: result.status,
            components: result.components,
            checked_at: result.checked_at,
            check_duration_ms: result.check_duration_ms,
        })
    }
}

/// 流健康检查结果
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct StreamHealthResult {
    /// 流标识符
    pub stream_id: String,
    /// 整体健康状态
    pub overall_status: HealthStatus,
    /// 组件健康状态
    pub components: std::collections::HashMap<String, ComponentHealth>,
    /// 检查时间
    pub checked_at: chrono::DateTime<chrono::Utc>,
    /// 检查耗时（毫秒）
    pub check_duration_ms: u64,
}

impl StreamHealthResult {
    /// 获取管道健康状态
    pub fn get_pipeline_health(&self, pipeline_id: &str) -> Option<&ComponentHealth> {
        self.components.get(&format!("pipeline.{}", pipeline_id))
    }

    /// 获取处理器健康状态
    pub fn get_processor_health(&self, processor_id: &str) -> Option<&ComponentHealth> {
        self.components.get(&format!("processor.{}", processor_id))
    }

    /// 检查流是否健康
    pub fn is_healthy(&self) -> bool {
        matches!(self.overall_status, HealthStatus::Healthy)
    }

    /// 检查是否有警告
    pub fn has_warnings(&self) -> bool {
        matches!(self.overall_status, HealthStatus::Warning) ||
        self.components.values().any(|c| matches!(c.status, HealthStatus::Warning))
    }

    /// 检查是否有严重问题
    pub fn has_critical_issues(&self) -> bool {
        matches!(self.overall_status, HealthStatus::Critical) ||
        self.components.values().any(|c| matches!(c.status, HealthStatus::Critical))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_processor_stats_success_rate() {
        let mut stats = ProcessorStats::default();
        
        stats.record_success(10.0);
        stats.record_success(20.0);
        stats.record_failure(15.0);
        
        assert_eq!(stats.total_messages, 3);
        assert_eq!(stats.successful_messages, 2);
        assert_eq!(stats.failed_messages, 1);
        assert_eq!(stats.success_rate(), 200.0 / 3.0);
        assert_eq!(stats.error_rate(), 100.0 / 3.0);
    }

    #[test]
    fn test_processor_stats_avg_time() {
        let mut stats = ProcessorStats::default();
        
        stats.record_success(10.0);
        stats.record_success(20.0);
        
        assert_eq!(stats.avg_processing_time_ms, 15.0);
    }

    #[tokio::test]
    async fn test_stream_pipeline_health_checker() {
        let checker = StreamPipelineHealthChecker::new(
            "test-pipeline".to_string(),
            Box::new(|| true),
        );

        let health = checker.check_health().await;
        assert_eq!(health.status, HealthStatus::Healthy);
        assert!(health.message.contains("test-pipeline"));
        assert!(health.message.contains("正在运行"));
    }

    #[tokio::test]
    async fn test_stream_processor_health_checker() {
        let stats = std::sync::Arc::new(tokio::sync::RwLock::new(ProcessorStats::default()));
        
        // 添加一些统计数据
        {
            let mut stats_guard = stats.write().await;
            stats_guard.record_success(10.0);
            stats_guard.record_success(15.0);
            stats_guard.record_failure(20.0);
        }

        let checker = StreamProcessorHealthChecker::new(
            "test-processor".to_string(),
            stats,
        );

        let health = checker.check_health().await;
        assert_eq!(health.status, HealthStatus::Warning); // 错误率 > 10%
        assert!(health.message.contains("test-processor"));
    }

    #[test]
    fn test_stream_health_result_methods() {
        let mut components = std::collections::HashMap::new();
        components.insert(
            "pipeline.test".to_string(),
            ComponentHealth {
                status: HealthStatus::Healthy,
                message: "OK".to_string(),
                details: std::collections::HashMap::new(),
                last_check: chrono::Utc::now(),
            },
        );

        let result = StreamHealthResult {
            stream_id: "test-stream".to_string(),
            overall_status: HealthStatus::Healthy,
            components,
            checked_at: chrono::Utc::now(),
            check_duration_ms: 100,
        };

        assert!(result.is_healthy());
        assert!(!result.has_warnings());
        assert!(!result.has_critical_issues());
        assert!(result.get_pipeline_health("test").is_some());
        assert!(result.get_processor_health("test").is_none());
    }
}